package com.atguigu.flink.chapter06.state.operator;

import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
/*
算子状态：
        广播状态（BroadcastState）

一个流，做成广播流（数据一般是一些配置数据）
另一个流，去connect 广播流

处理广播流数据的时候，把数据存入到广播状态，处理数据流数据的时候，从广播状态读取配置数据
 */
public class BroadCastStateDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        DataStreamSource<String> dataStream = env.socketTextStream("hadoop102", 8888);

        DataStreamSource<String> configStream = env.socketTextStream("hadoop102", 9999);

        MapStateDescriptor<String, String> desc = new MapStateDescriptor<>("bcState", String.class, String.class);

        //1、把配置信息做成广播流
        BroadcastStream<String> bcStream = configStream.broadcast(desc);

        //2、让数据流去connect 广播流
        BroadcastConnectedStream<String, String> dataBcStream = dataStream.connect(bcStream);

        //3、处理connect 之后的数据
        dataBcStream.process(new BroadcastProcessFunction<String, String, String>() {
            //处理数据流中的元素
            //从上下文中读取广播状态的值，根据值进行具体的处理
            @Override
            public void processElement(String value,
                                       ReadOnlyContext ctx,
                                       Collector<String> out) throws Exception {
                ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(desc);
                String v = state.get("switch");

                if ("1".equals(v)){
                    System.out.println("使用 1 号逻辑来处理数据："  + value);
                }else if ("2".equals(v)){
                    System.out.println("使用 2 号逻辑来处理数据："  + value);
                }else {
                    System.out.println("使用 default 号逻辑来处理数据："  + value);
                }
            }

            //处理广播流中的元素
            //把数据存入广播状态中
            @Override
            public void processBroadcastElement(String value,
                                                Context ctx,
                                                Collector<String> out) throws Exception {
                System.out.println(value + "....");
                BroadcastState<String, String> state = ctx.getBroadcastState(desc);

                state.put("switch",value);
            }
        })
        .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}




/*
一个流做成广播流（数据是一些配置数据）
另一个流去connect广播流

处理广播流数据的时候，把数据存入广播状态，处理数据流数据的时候，从广播流读取配置数据
 */
//public class BroadCastStateDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        DataStreamSource<String> dataStream = env.socketTextStream("hadoop102", 8888);
//        DataStreamSource<String> configStream = env.socketTextStream("hadoop102", 9999);
//
//        MapStateDescriptor<String, String> desc = new MapStateDescriptor<>("bcState", String.class, String.class);
//
//        //1、把配置信息做成广播流
//        BroadcastStream<String> bcStream = configStream.broadcast(desc);
//
//        //2、让数据流去connect广播流
//        BroadcastConnectedStream<String, String> dataBcStream = dataStream.connect(bcStream);
//
//        //3、处理connect之后的数据
//        dataBcStream.process(new BroadcastProcessFunction<String, String, String>() {
//            //处理数据流中的元素
//            //从上下文中读取广播状态的值，根据值进行具体的处理
//            @Override
//            public void processElement(String value,
//                                       ReadOnlyContext ctx,
//                                       Collector<String> out) throws Exception {
//                ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                String v = state.get("switch");
//                if ("1".equals(v)){
//                    System.out.println("使用 1 号逻辑来处理数据："  + value);
//                }else if ("2".equals(v)){
//                    System.out.println("使用 2 号逻辑来处理数据："  + value);
//                }else {
//                    System.out.println("使用 default 号逻辑来处理数据："  + value);
//                }
//
//            }
//
//            //处理广播流中数据
//            //把数据存入广播状态中
//            @Override
//            public void processBroadcastElement(String value,
//                                                Context ctx,
//                                                Collector<String> out) throws Exception {
//                System.out.println("value......");
//                BroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                state.put("switch",value);
//            }
//        })
//                .print();
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}




/*
一个流，做成广播流（数据是一些配置数据）
另一个流，去connect 广播流

处理广播流数据的时候，把数据存入广播状态，处理数据流数据的时候，从广播状态读取配置信息
 */
//public class BroadCastStateDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        DataStreamSource<String> dataStream = env.socketTextStream("hadoop102", 8888);
//        DataStreamSource<String> configStream = env.socketTextStream("hadoop102", 9999);
//        MapStateDescriptor<String, String> desc = new MapStateDescriptor<>("bcState", String.class, String.class);
//
//        //1、把配置信息做成广播流
//        BroadcastStream<String> bcStream = configStream.broadcast(desc);
//
//        //2、让数据流去connect广播流
//        BroadcastConnectedStream<String, String> dataBcStream = dataStream.connect(bcStream);
//
//        //3、处理connect 之后的数据
//        dataBcStream.process(new BroadcastProcessFunction<String, String, String>() {
//            //处理数据流中的元素
//            //从配置信息中读取广播状态的值，根据值进行具体的处理
//            @Override
//            public void processElement(String value,
//                                       ReadOnlyContext ctx,
//                                       Collector<String> out) throws Exception {
//                ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                String s = state.get("switch");
//
//                if ("1".equals(s)){
//                    System.out.println("使用 1 号逻辑来处理数据："  + value);
//                }else if ("2".equals(s)){
//                    System.out.println("使用 2 号逻辑来处理数据："  + value);
//                }else {
//                    System.out.println("使用 default 号逻辑来处理数据："  + value);
//                }
//            }
//
//            //处理广播流中的元素
//            //把数据写入广播状态中
//            @Override
//            public void processBroadcastElement(String value,
//                                                Context ctx,
//                                                Collector<String> out) throws Exception {
//                System.out.println("value............");
//                BroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                state.put("switch",value);
//            }
//        })
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}



/*
一个流，做成广播流（数据是一些配置数据）
另一个流去connect广播流

处理广播流数据的时候，把数据存入广播状态中，处理数据流数据的时候，从广播状态读取配置数据
 */
//public class BroadCastStateDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        DataStreamSource<String> dataStream = env.socketTextStream("hadoop102", 8888);
//        DataStreamSource<String> configStream = env.socketTextStream("hadoop102", 9999);
//        MapStateDescriptor<String, String> desc = new MapStateDescriptor<>("bcState", String.class, String.class);
//
//        //1、把配置信息做成广播流
//        BroadcastStream<String> bcStream = configStream.broadcast(desc);
//
//        //2、让数据流去connect广播流
//        BroadcastConnectedStream<String, String> dataBcStream = dataStream.connect(bcStream);
//
//        //3、处理connect之后的数据
//        dataBcStream.process(new BroadcastProcessFunction<String, String, String>() {
//            //处理数据流中元素
//            //从配置信息中读取广播状态的值，根据值进行具体的处理
//            @Override
//            public void processElement(String value,
//                                       ReadOnlyContext ctx,
//                                       Collector<String> out) throws Exception {
//                ReadOnlyBroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                String v = state.get("switch");
//
//                if ("1".equals(v)){
//                    System.out.println("使用 1 号逻辑来处理数据："  + value);
//                }else if ("2".equals(v)){
//                    System.out.println("使用 2 号逻辑来处理数据："  + value);
//                }else {
//                    System.out.println("使用 default 号逻辑来处理数据："  + value);
//                }
//            }
//
//            //处理广播流中数据
//            //把数据存入广播状态中
//            @Override
//            public void processBroadcastElement(String value,
//                                                Context ctx,
//                                                Collector<String> out) throws Exception {
//                System.out.println("value..............");
//                BroadcastState<String, String> state = ctx.getBroadcastState(desc);
//                state.put("switch",value);
//            }
//        })
//                .print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}